package com.iacrqq.sedaf.thread;

import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import com.iacrqq.sedaf.common.Controllable;
import com.iacrqq.sedaf.config.Configuration;

/**
 * 
 * @author sihai
 * 
 */
public class DefaultThreadPool implements ThreadPool
{

	private static final Log logger = LogFactory.getLog(DefaultThreadPool.class);

	private String name;
	private Runnable runnable;
	private Configuration configuration;
	private int minThreads;
	private int maxThreads;
	private ThreadGroup threadGroup;

	private List<ThreadWorker> workerList;
	private List<ThreadWorker> idleWorkerList;

	private ThreadFactory threadFactory;

	public DefaultThreadPool(final String name, Runnable runnable)
	{
		this.name = name;
		this.runnable = runnable;
	}

	@Override
	public synchronized int idle(int num)
	{
		int n = 0;
		if(_get_working_thread_count_() == 1)
		{
			logger.warn(new StringBuilder(name).append(" try to be idle last working thread."));
			return n;
		}
		
		for(ThreadWorker worker : workerList)
		{
			if(!worker.isIdle())
			{
				worker.idle(true);
				workerList.remove(worker);
				idleWorkerList.add(worker);
				n++;
			}
		}
		
		return n;
	}
	
	private int _get_working_thread_count_()
	{
		return workerList.size() - idleWorkerList.size();
	}

	@Override
	public synchronized int addThreads(int num)
	{
		int num4Add = Math.min(num, maxThreads - size());
		
		Iterator<ThreadWorker> iterator = idleWorkerList.iterator();
		while (iterator.hasNext())
		{
			ThreadWorker worker = iterator.next();
			iterator.remove();
			worker.setRunnable(runnable);
			worker.idle(false);
			num4Add -= 1;
		}

		if (num4Add > 0)
		{
			_add_worker_(num4Add);
		}
		
		return num;
	}

	private void _add_worker_(int num)
	{
		for (int i = 0; i < num; i++)
		{
			ThreadWorker worker = new ThreadWorker(runnable, false);
			Thread thread = threadFactory.newThread(worker);
			worker.setThread(thread);
			workerList.add(worker);
			worker.start();
		}
	}

	@Override
	public synchronized int removeIdleThreads()
	{
		int n = 0;
		
		Iterator<ThreadWorker> iterator = idleWorkerList.iterator();
		while (iterator.hasNext())
		{
			ThreadWorker worker = iterator.next();
			iterator.remove();
			worker.stop();
			n++;
		}
		
		return n;
	}

	@Override
	public synchronized int removeThreads(int num)
	{
		// remove idle first
		int n = removeIdleThreads();
		if(_get_working_thread_count_() == 1)
		{
			logger.warn(new StringBuilder(name).append(" try to be removed last thread."));
		}
		
		return n;
	}

	@Override
	public void setRunnable(Runnable runnable)
	{
		this.runnable = runnable;
	}

	@Override
	public String getName()
	{
		return name;
	}

	@Override
	public void setName(String name)
	{
		this.name = name;
	}

	@Override
	public void init(Configuration config) throws Exception
	{
		this.configuration = config;
		this.minThreads = configuration.getInt(new StringBuilder(name).append(".minThreads").toString(), 1);
		this.maxThreads = configuration.getInt(new StringBuilder(name).append(".maxThreads").toString(), 1);

		workerList = new LinkedList<ThreadWorker>();
		idleWorkerList = new LinkedList<ThreadWorker>();

		threadGroup = new ThreadGroup(new StringBuilder(name).append("-ThreadGroup").toString());

		threadFactory = new ThreadFactory()
		{
			public Thread newThread(Runnable r)
			{
				return new Thread(threadGroup, r, new StringBuilder(name).append("-ThreadPool-Thread-").append(size())
						.toString());
			}
		};

		addThreads(this.minThreads);

		logger.debug(new StringBuilder("ThreadPool:").append(name).append(" initilized, minThreads:")
				.append(minThreads).append(",maxThreads:").append(maxThreads).toString());
	}

	@Override
	public void destroy() throws Exception
	{

	}

	@Override
	public void pause()
	{

	}

	@Override
	public void start()
	{

	}

	@Override
	public void stop()
	{
		/*
		 * synchronized(_workerList_lock_) { Iterator<ThreadWorker> iterator =
		 * workerList.iterator(); while(iterator.hasNext()) { ThreadWorker
		 * worker = iterator.next(); iterator.remove(); worker.stop(); } }
		 */
		threadGroup.interrupt();
	}

	@Override
	public synchronized int size()
	{
		return workerList.size();
	}

	@Override
	public synchronized int idleSize()
	{
		return this.idleWorkerList.size();
	}

	private class ThreadWorker implements Runnable, Controllable
	{

		private AtomicReference<Runnable> runnable;
		private volatile boolean isIdle = true;
		private Thread thread;

		public ThreadWorker(Runnable runnable, boolean isIdle)
		{
			this.runnable = new AtomicReference<Runnable>();
			this.runnable.set(runnable);
			this.isIdle = isIdle;
		}

		@Override
		public void run()
		{
			while (!Thread.currentThread().isInterrupted())
			{
				while (isIdle)
				{
					if (Thread.currentThread().isInterrupted())
					{
						break;
					}
					logger.debug(new StringBuilder("Thread:").append(Thread.currentThread().getName()).append(
							", idle, yield."));
					Thread.yield();
				}

				if (!Thread.currentThread().isInterrupted())
				{
					logger.debug(new StringBuilder("Thread:").append(Thread.currentThread().getName()).append(
							", call stage run method"));
					//
					runnable.get().run();
				}
			}
		}

		@Override
		public void pause()
		{

		}

		@Override
		public void start()
		{
			thread.start();
		}

		@Override
		public void stop()
		{
			thread.interrupt();
		}

		public void idle(boolean isIdle)
		{
			this.isIdle = isIdle;
		}
		
		public boolean isIdle()
		{
			return this.isIdle;
		}

		public void setRunnable(Runnable runnable)
		{
			this.runnable.set(runnable);
		}

		public void setThread(Thread thread)
		{
			this.thread = thread;
		}
	}

	public static void main(String[] args)
	{
		try
		{
			DefaultThreadPool pool = new DefaultThreadPool("test", new Tester());
			pool.init(null);
			Thread.sleep(60000);
			pool.stop();
		}
		catch (InterruptedException e)
		{
			e.printStackTrace();
		}
		catch (Exception e)
		{
			e.printStackTrace();
		}
	}

	private static class Tester implements Runnable
	{

		@Override
		public void run()
		{
			System.out.println(Thread.currentThread().getName());
		}
	}
}
